package com.fudan.run.trans;

import java.io.Serializable;
import java.util.Map;

import org.apache.spark.api.java.JavaRDD;
import org.codehaus.janino.Java;

import com.fudan.cfg.RunConfiguration;
import com.fudan.cfg.base.TransDef;
import com.fudan.run.JobRunner;
import com.fudan.run.VariableRddMap;
import com.fudan.run.ctx.TransContext;

public abstract class TransHandler<T extends TransDef> implements Serializable{

	public static TransHandler<? super TransDef> build(TransDef transNode,RunConfiguration runConfiguration) {
		String name = transNode.getName();
		String type = transNode.getType();
		TransContext transContext = runConfiguration.getTransContext();
		Class<? extends TransHandler> handler = transContext.getHandler(type);
		if(handler != null) {
			try {
				return  handler.newInstance();
			} catch (InstantiationException | IllegalAccessException e) {
				throw new RuntimeException(String.format("构建trans的处理器失败，name:%s,type:%s",name,type),e);
			}
		}
//		if("combine".equals(type)) {
//			return new CombineTrans();
//		}
//		if("test".equals(type)) {
//			return new TestTrans();
//		}

		throw new RuntimeException(String.format("没有trans的处理器，name:%s,type:%s",name,type));
	}

	public abstract void trans(JobRunner jobRunner,T transNode,VariableRddMap variableMap);
}
